package org.yhx.learning.jukebox;/*
 * Copyright (C), 2024, bigmoon
 * FileName:
 * Author:   bigmoon
 * Date:     2024/9/6 14:49
 * Description: //模块目的、功能描述
 * History: //修改记录
 * &lt;author&gt;      &lt;time&gt;      &lt;version&gt;    &lt;desc&gt;
 * 修改人姓名             修改时间            版本号                  描述
 */

import io.vertx.core.AbstractVerticle;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.eventbus.EventBus;
import io.vertx.core.eventbus.Message;
import io.vertx.core.file.AsyncFile;
import io.vertx.core.file.OpenOptions;
import io.vertx.core.http.HttpServerRequest;
import io.vertx.core.http.HttpServerResponse;
import io.vertx.core.json.JsonArray;
import io.vertx.core.json.JsonObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayDeque;
import java.util.HashSet;
import java.util.Queue;
import java.util.Set;

public class JukeBox extends AbstractVerticle {

    private static final Logger LOGGER = LoggerFactory.getLogger(JukeBox.class);

    private static final String path_prefix = "tracks";


    private enum Status {PLAYING, PAUSED}

    private Status currentMode = Status.PAUSED;
    private final Queue<String> playList = new ArrayDeque<>();
    private final Set<HttpServerResponse> streams = new HashSet<>();
    private long currentPosition;
    private AsyncFile currentFile;

    @Override
    public void start() throws Exception {

        EventBus eventBus = vertx.eventBus();
        eventBus.consumer("jukebox.list", this::list);
        eventBus.consumer("jukebox.schedule", this::schedule);
        eventBus.consumer("jukebox.play", this::play);
        eventBus.consumer("jukebox.pause", this::pause);

        vertx.createHttpServer().requestHandler(this::httpHandler).listen(8080);
        vertx.setPeriodic(200, this::streamChunk);

    }

    //    获取所有mp3文件名称并返回
    private void list(Message msg) {
        vertx.fileSystem().readDir(path_prefix, ".*mp3$", ar -> {

            if (ar.succeeded()) {
                JsonArray objects = new JsonArray(ar.result());
                msg.reply(objects);
            } else {
                LOGGER.error("获取MP3文件列表异常");
                msg.fail(500, ar.cause().getMessage());
            }
        });

    }

    //    播放列表
    private void schedule(Message<JsonObject> msg) {

        String file = msg.body().getString("file");

        if (this.playList.isEmpty() && this.currentMode == Status.PAUSED) {
            this.currentMode = Status.PLAYING;
        }
        this.playList.offer(file);
    }

    private void play(Message msg) {
        this.currentMode = Status.PLAYING;

    }

    private void pause(Message msg) {
        this.currentMode = Status.PAUSED;
    }

    private void httpHandler(HttpServerRequest request) {

        if (request.path().equals("/")) {
            openAudioStream(request);

        } else if (request.path().startsWith("/download")) {
            String name = request.getParam("name");
            download(name, request);
        } else {
            request.response().setStatusCode(404).end();
        }
    }

    private void openAudioStream(HttpServerRequest request) {

        HttpServerResponse response = request.response();
        response.setChunked(true);
        response.putHeader("Context-type", "audio/mpeg");
        streams.add(response);
        response.endHandler(Void -> {
            this.streams.remove(response);
            LOGGER.info("A stream left");
        });
    }

    private void download(String filename, HttpServerRequest request) {

        OpenOptions openOptions = new OpenOptions().setRead(true);
        HttpServerResponse response = request.response();
        response.setChunked(true);
        response.putHeader("Context-type", "audio/mpeg");
        String filePath = path_prefix + "/" + filename + ".mp3";

        if (!vertx.fileSystem().existsBlocking(filePath)) {
            response.setStatusCode(404).end();
            return;
        }

        vertx.fileSystem().open(filePath, openOptions, ar -> {

            if (ar.succeeded()) {

                ar.result().handler(buffer -> {
                    response.write(buffer);
                    if (response.writeQueueFull()) {
                        ar.result().pause();
                    }
                    //当写入队列消耗一半以上时，再次启动文件读取缓存
                    response.drainHandler(v -> ar.result().resume());
                }).endHandler(Void -> {
                    response.end();
                    LOGGER.info("文件{} 全部读取完成", filename);
                });
            } else {
                response.setStatusCode(500).setStatusMessage(ar.cause().getMessage()).end();
                LOGGER.error("文件{} 读取发生异常",filename, ar.cause());
            }
        });
    }

    private void streamChunk(Long id) {

        if (this.currentMode == Status.PAUSED) {
            return;
        }

        if (null == this.playList || this.playList.isEmpty()) {
            return;
        }

        if (this.currentFile == null) {

            OpenOptions options = new OpenOptions();
            options.setRead(true);
            vertx.fileSystem().open(path_prefix + this.playList.poll(), options, ar -> {
                if (ar.succeeded()) {
                    this.currentFile = ar.result();
                }
            });
        }

        this.currentFile.read(Buffer.buffer(4096), 0, this.currentPosition, 4096, ar -> {

            if (ar.succeeded()) {

                if (ar.result().length() <= 0) {
                    this.currentFile.close().onSuccess(v -> this.currentFile = null);
                    this.currentPosition = 0;
                    this.currentMode = Status.PAUSED;
                }

                this.currentPosition += ar.result().length();

                this.streams.forEach(stream -> {

                    if (!stream.writeQueueFull()) {
                        //重点 缓存流不能重复使用，必须copy新
                        stream.write(ar.result().copy());
                    }
                });
            } else {

                LOGGER.error("异常", ar.cause());
            }

        });


    }
}
